package my.rpc.client;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import my.rpc.common.bean.RpcRequest;
import my.rpc.common.bean.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * RPC客户端Handler
 * @author fengbo
 * @date 2017/11/2
 */
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Logger log = LoggerFactory.getLogger(RpcClientHandler.class);

    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private RpcResponse response;
    private volatile Channel channel;

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) {
        // 接受服务器返回的数据（在解码器执行后执行）
        try {
            lock.lock();
            this.response = response;
            condition.signal();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("client caught exception", cause);
        ctx.close();
    }

    RpcResponse sendRequest(RpcRequest request) throws InterruptedException {
        try {
            // TODO 修改为非阻塞模式
            lock.lock();
            channel.writeAndFlush(request);
            condition.await();
            return response;
        } finally {
            lock.unlock();
        }
    }
}
